package com.example.base;

import cn.hutool.core.lang.UUID;
import com.example.entity.Order;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
import org.apache.flink.connector.datagen.source.DataGeneratorSource;
import org.apache.flink.connector.datagen.source.GeneratorFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.Arrays;
import java.util.List;
import java.util.Random;


/**
 * Author wangJinLong
 * Date 2025/8/12 16:59
 **/
public class FlinkStreamApp {

    private static List<String> faceList = Arrays.asList("14901", "24902", "34903", "44903", "54903");
    private static List<String> siteList = Arrays.asList("05350101", "05350104", "05350122", "00004", "00002");


    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStream<Order> mapSource = FlinkStreamSource.getMapSource(env);

        KeyedStream<Order, String> keyedStream = mapSource.keyBy(Order::getOrderId);

        // 开一个10秒的窗口
        WindowedStream<Order, String, TimeWindow> windowedStream = keyedStream.window(TumblingEventTimeWindows.of(Duration.ofSeconds(10)));

        DataStream<Order> processStream = windowedStream.process(new ProcessWindowFunction<Order, Order, String, TimeWindow>() {
            @Override
            public void process(String s, Context context, Iterable<Order> elements, Collector<Order> out) throws Exception {
                int num = 0;
                for (Order order : elements) {
                    num++;
                }
                Order order = new Order();
                order.setOrderId(s);
                order.setAmount(num);
                out.collect(order);
            }
        });

        processStream.print();

        env.execute();
    }

    private static void windowApply() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStream<Order> mapSource = FlinkStreamSource.getMapSource(env);

        KeyedStream<Order, String> keyedStream = mapSource.keyBy(Order::getOrderId);

        // 开一个10秒的窗口
        WindowedStream<Order, String, TimeWindow> windowedStream = keyedStream.window(TumblingEventTimeWindows.of(Duration.ofSeconds(10)));

        // 计算窗口内orderId的生成数量
        DataStream<Order> apply = windowedStream.apply(new WindowFunction<Order, Order, String, TimeWindow>() {
            @Override
            public void apply(String s, TimeWindow window, Iterable<Order> orders, Collector<Order> out) throws Exception {
                int num = 0;
                for (Order order : orders) {
                    num++;
                }
                Order order = new Order();
                order.setOrderId(s);
                order.setAmount(num);
                out.collect(order);
            }
        });

        apply.print();

        env.execute();
    }

    private static void reduceStream() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStream<Order> mapSource = FlinkStreamSource.getMapSource(env);

        KeyedStream<Order, String> keyedStream = mapSource.keyBy(Order::getOrderId);

        DataStream<Order> reduceStream = keyedStream.reduce(new ReduceFunction<Order>() {
            @Override
            public Order reduce(Order value1, Order value2) throws Exception {
                Order order = new Order();
                order.setOrderId(value1.getOrderId());
                order.setAmount(value1.getAmount() * value2.getAmount());
                order.setTotalPrice(value1.getTotalPrice());
                order.setOrderTime(value1.getOrderTime());
                return order;
            }
        });

        reduceStream.print();

        env.execute();
    }




    private static void flatmap() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(3);

        GeneratorFunction<Long, String> generatorFunction = value -> {
            Random random = new Random();
            LocalDateTime localDateTime = LocalDateTime.now();
            long epochMilli = localDateTime.toInstant(ZoneOffset.systemDefault().getRules().getOffset(localDateTime)).toEpochMilli();
            String str = UUID.randomUUID()+ "," + random.nextInt(10) + "," + random.nextDouble(10) + "," + epochMilli;
            return str;
        };

        DataGeneratorSource<String> source = new DataGeneratorSource<>(
                generatorFunction,
                60L,
                RateLimiterStrategy.perSecond(1),
                Types.STRING
        );

        DataStream<String> mapSource = env.fromSource(
                source,
                WatermarkStrategy.noWatermarks()
                , "map source"
        );

        DataStream<String> flatMap = mapSource.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                String[] strings = value.split(",");
                for (String string : strings) {
                    out.collect(string);
                }
            }
        });

        flatMap.print();

        env.execute();
    }

    private static void mapStream() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(3);

        DataStream<Order> mapSource = FlinkStreamSource.getMapSource(env);

        DataStream<Order> mappedStream = mapSource.map(new MapFunction<Order, Order>() {
            @Override
            public Order map(Order value) throws Exception {
                value.setAmount(value.getAmount() * 10);
                return value;
            }
        });

        mappedStream.print();

        env.execute();
    }
}
